package com.rizzo.analytics.services.web.delegator;


import com.google.common.collect.*;
import com.rizzo.analytics.domain.CountResult;
import com.rizzo.analytics.domain.application.Applications;
import com.rizzo.analytics.domain.errors.Error;
import com.rizzo.analytics.domain.errors.Errors;
import com.rizzo.analytics.domain.info.Info;
import com.rizzo.analytics.domain.info.Instance;
import com.rizzo.analytics.domain.notification.Notification;
import com.rizzo.analytics.domain.notification.Notifications;
import com.rizzo.analytics.domain.notification.NotifiedService;
import com.rizzo.analytics.domain.performance.CallInfo;
import com.rizzo.analytics.domain.performance.MonitoredService;
import com.rizzo.analytics.domain.performance.Performance;
import com.rizzo.analytics.domain.quota.Quota;
import com.rizzo.analytics.domain.quota.Quotas;
import com.rizzo.analytics.domain.trend.Trend;
import com.rizzo.analytics.domain.trend.TrendPoint;
import com.rizzo.analytics.domain.trend.TrendType;
import com.rizzo.analytics.domain.trend.TrendedService;
import com.rizzo.analytics.services.web.controller.v1.error.ErrorsNotFoundException;
import com.rizzo.analytics.services.web.controller.v1.error.InfoNotFoundException;
import com.rizzo.analytics.services.web.controller.v1.error.QuotasNotFoundException;
import com.rizzo.infrastructure.domain.KeyType;
import com.rizzo.infrastructure.redis.config.RedisConfigManager;
import com.rizzo.infrastructure.redis.key.RedisKey;
import com.rizzo.infrastructure.redis.key.RedisKeyBuilder;
import com.rizzo.infrastructure.redis.timeseries.TimeStep;
import com.rizzo.infrastructure.util.MatchingUtil;
import org.joda.time.DateTime;
import org.joda.time.Period;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.BoundZSetOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.StringUtils;

import java.util.BitSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class AnalyticsDelegatorImpl extends RizzoDelegator implements AnalyticsDelegator {

    private static final Logger LOGGER = LoggerFactory.getLogger(AnalyticsDelegatorImpl.class);

    private RedisOperations<String, String> stringRedisOperations;
    private RedisOperations<String, BitSet> bitmapRedisOperations;
    private RedisOperations<RedisKey, String> redisKeyRedisOperations;
    private RedisOperations<RedisKey, Long> longRedisOperations;

    private RedisConfigManager redisConfigManager;

    @Cacheable(value = "applications", key = "#user")
    public Applications getApplications(String user) {
        Set<String> keys = stringRedisOperations.keys(RedisKey.REDIS_WILDCARD);
        Set<String> applicationString = Sets.newTreeSet();
        for (String key : keys) {
            RedisKey redisKey = RedisKey.deSerialize(key);
            if (redisKey.getData() != null && redisKey.getData().contains(user)) {
                applicationString.add(redisKey.getApplication());
            }
        }
        return new Applications(applicationString);
    }

    @Cacheable(value = "info", key = "#user.concat('-').concat(#application).concat('-').concat(#keyType).concat('-').concat(#category)" +
            ".concat('-').concat(#instanceName).concat('-').concat(#instanceIndex)")
    public Info getInfo(String user, String application, KeyType keyType, String category, String instanceName, String instanceIndex) {
        Info info = new Info();
        String keysCriteria;
        RedisKey performanceCriteria = new RedisKeyBuilder().key(application).type(KeyType.PERFORMANCE)
                .instanceName(StringUtils.hasText(instanceName) ? instanceName : RedisKey.REDIS_WILDCARD)
                .instanceIndex(StringUtils.hasText(instanceIndex) ? instanceIndex : RedisKey.REDIS_WILDCARD)
                .category(StringUtils.hasText(category) ? category : RedisKey.REDIS_WILDCARD)
                .data(user + RedisKey.REDIS_WILDCARD).build();
        RedisKey errorCriteria = new RedisKeyBuilder().key(application).type(KeyType.ERROR)
                .data(user).build();
        RedisKey notificationCriteria = new RedisKeyBuilder().key(application).type(KeyType.NOTIFICATION)
                .category(StringUtils.hasText(category) ? category : RedisKey.REDIS_WILDCARD)
                .data(user).data(RedisKey.REDIS_WILDCARD).build();
        RedisKey quotaCriteria = new RedisKeyBuilder().key(application).type(KeyType.QUOTA)
                .category((StringUtils.hasText(category) ? category : RedisKey.REDIS_WILDCARD))
                .data(user).build();
        RedisKey cacheCriteria = new RedisKeyBuilder().key(application).type(KeyType.CACHE)
                .data(RedisKey.REDIS_WILDCARD).data(RedisKey.REDIS_WILDCARD + user + RedisKey.REDIS_WILDCARD).build();
        Map<KeyType, RedisKey> criteria = ImmutableMap.of(KeyType.PERFORMANCE, performanceCriteria,
                KeyType.ERROR, errorCriteria, KeyType.NOTIFICATION, notificationCriteria, KeyType.QUOTA, quotaCriteria,
                KeyType.CACHE, cacheCriteria);
        if (keyType != KeyType.ALL) {
            keysCriteria = criteria.get(keyType).serialize();
            info.merge(getInfo(keysCriteria));
        } else {
            for (Map.Entry<KeyType, RedisKey> keyTypeRedisKeyEntry : criteria.entrySet()) {
                RedisKey redisKey = keyTypeRedisKeyEntry.getValue();
                info.merge(getInfo(redisKey.serialize()));
            }
        }
        if (info.isEmpty()) {
            throw new InfoNotFoundException(40444, "service.info.get.not.found", new Object[]{application});
        }

        return info;
    }

    @Cacheable(value = "performance",
            key = "#user.concat('-').concat(#application).concat('-').concat(#instanceName).concat('-').concat(#instanceIndex).concat('-')" +
                    ".concat(#category).concat('-').concat(#limit).concat('-').concat(#offset)" +
                    ".concat('-').concat(#callSort)")
    public Performance getPerformance(final String user, final String application, final String instanceName,
                                      final String instanceIndex, final String category, final long limit, final long offset, final CallInfo.CallSort callSort) {
        Performance performance = new Performance();

        // make a key that matches all the asked for criteria
        RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.PERFORMANCE)
                .instanceName(StringUtils.hasText(instanceName) ? instanceName : RedisKey.REDIS_WILDCARD)
                .instanceIndex(StringUtils.hasText(instanceIndex) ? instanceIndex : RedisKey.REDIS_WILDCARD)
                .category(StringUtils.hasText(category) ? category : RedisKey.REDIS_WILDCARD)
                .data(StringUtils.hasText(user) ? user : RedisKey.ANONYMOUS)
                .data(RedisKey.REDIS_WILDCARD).build();

        Set<String> keys = stringRedisOperations.keys(redisKey.serialize()); // TODO   test if keys-method returns null when no keys were found
        for (final String key : keys) {
            final RedisKey rKey = RedisKey.deSerialize(key);
            final long max = ((offset + limit) * TimeStep.DATA_POINT_LENGTH);
            final long min = (offset == 0) ? 1 : (offset * TimeStep.DATA_POINT_LENGTH) + 1;
            String timeStep = rKey.getSecondData();
            if (!TrendType.isTrendType(timeStep)) {
                MonitoredService foundMonitoredService = performance.getMonitoredService(rKey.getCategory(), rKey.getInstanceName(), rKey.getInstanceIndex());
                Set<CallInfo> callInfos = stringRedisOperations.execute(new RedisCallback<Set<CallInfo>>() {
                    @Override
                    public Set<CallInfo> doInRedis(RedisConnection connection) throws DataAccessException {
                        // we have to make sure we query in datapoints of 19 length starting from the back
                        byte[] range = connection.getRange(key.getBytes(), -max, -min);
                        if (range != null && StringUtils.hasLength(new String(range)) && range.length != 1) {
                            return CallInfo.parseRange(new String(range), callSort);
                        } else {
                            return Sets.newHashSet();
                        }
                    }
                });
                long callTotal = stringRedisOperations.opsForValue().size(key) / TimeStep.DATA_POINT_LENGTH;
                if (!callInfos.isEmpty()) {
                    if (foundMonitoredService == null) {
                        MonitoredService monitoredService = new MonitoredService(rKey.getCategory(), rKey.getInstanceName(), rKey.getInstanceIndex(),
                                callTotal, callInfos);
                        performance.addMonitoredService(monitoredService);
                    } else {
                        foundMonitoredService.getCalls().addAll(callInfos);
                        foundMonitoredService.addCallTotal(callTotal);
                    }
                }
            }
        }
        return performance;
    }

    @Override
    public CountResult savePerformance(String user, String application, Performance performance) {
        Long savedCalls = 0L;
        final Period performanceTimeStep = TimeStep.parsePeriod(redisConfigManager.getUserConfig(user, application, "performanceTimeStep"));
        final String performanceDurationZeroPaddings = redisConfigManager.getSystemConfig("performanceDurationZeroPaddings");
        if (performance != null && performance.getMonitoredServices() != null) {
            for (MonitoredService monitoredService : performance.getMonitoredServices()) {
                for (CallInfo callInfo : monitoredService.getCalls()) {
                    final String timeStep = TimeStep.encodeTimeStep(new DateTime(callInfo.getTime()), performanceTimeStep);
                    final String now = TimeStep.encodeTimeStep(DateTime.now().minus(performanceTimeStep), performanceTimeStep);
                    final String next = TimeStep.encodeTimeStep(DateTime.now(), performanceTimeStep);
                    RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.PERFORMANCE)
                            .instanceName(monitoredService.getInstanceName())
                            .instanceIndex(monitoredService.getInstanceIndex())
                            .category(monitoredService.getService())
                            .data(user).data(timeStep).build();
                    // append the callinfo's into the stringRedis appending value
                    String logStatementTimestamp = String.valueOf(callInfo.getTime().getMillis());
                    String durationPadded = String.format("%0" + performanceDurationZeroPaddings + "d", callInfo.getDuration());
                    // provided data must lie in the running timestep (or next one) otherwise we ignore it in fear of overriding trend data!
                    if (timeStep.equals(now) || timeStep.equals(next)) {
                        savedCalls++;
                        stringRedisOperations.opsForValue().append(redisKey.serialize(), logStatementTimestamp + durationPadded);
                    } else {
                        LOGGER.warn("User posted old performance data so ignoring it!");
                    }
                }
            }
        }
        return new CountResult(savedCalls);
    }

    @Cacheable(value = "trend",
            key = "#user.concat('-').concat(#application).concat('-').concat(#instanceName).concat('-').concat(#instanceIndex).concat('-')" +
                    ".concat(#category).concat('-').concat(#limit).concat('-').concat(#offset).concat('-').concat(#trendType)")
    public Trend getTrend(final String user, final String application, final String instanceName, final String instanceIndex,
                          final String category, final long limit, final long offset, final TrendType trendType) {
        // make a key that matches all the asked for criteria
        Trend trend = new Trend();
        RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.PERFORMANCE)
                .instanceName(StringUtils.hasText(instanceName) ? instanceName : RedisKey.REDIS_WILDCARD)
                .instanceIndex(StringUtils.hasText(instanceIndex) ? instanceIndex : RedisKey.REDIS_WILDCARD)
                .category(StringUtils.hasText(category) ? category : RedisKey.REDIS_WILDCARD)
                .data(StringUtils.hasText(user) ? user : RedisKey.ANONYMOUS)
                .data((trendType != TrendType.ALL) ? trendType.getPrefix() : RedisKey.REDIS_WILDCARD).build();

        Set<String> keys = stringRedisOperations.keys(redisKey.serialize()); // TODO  test if keys-method returns null when no keys were found
        for (final String key : keys) {
            final RedisKey rKey = RedisKey.deSerialize(key);
            String suffix = rKey.getSecondData();
            if (TrendType.isTrendType(suffix)) {
                BoundZSetOperations<RedisKey, String> zSetOperations = this.redisKeyRedisOperations.boundZSetOps(rKey);
                Set<ZSetOperations.TypedTuple<String>> typedTuples = zSetOperations.reverseRangeWithScores(offset, offset + limit);
                String service = rKey.getCategory();
                TrendedService foundTrendedService = trend.getTrendedService(service);
                if (typedTuples != null) {
                    if (foundTrendedService != null) {
                        for (ZSetOperations.TypedTuple<String> tuple : typedTuples) {
                            foundTrendedService.trendData()
                                    .put(TrendType.fromPrefix(suffix), new TrendPoint(new DateTime(Long.parseLong(tuple.getValue())), tuple.getScore()));
                        }
                    } else {
                        TrendedService trendedService = convertTrend(rKey.getInstanceName(), rKey.getInstanceIndex(),
                                service, TrendType.fromPrefix(suffix), typedTuples);
                        trend.addTrendedService(trendedService);
                    }
                }
            }
        }
        return trend;
    }

    @Cacheable(value = "errors",
            key = "#user.concat('-').concat(#application).concat('-').concat(#code).concat('-').concat(#limit).concat('-').concat(#offset)")
    public Errors getErrors(String user, String application, String code, long limit, long offset) throws ErrorsNotFoundException {
        RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.ERROR).data(user).build();
        BoundZSetOperations<RedisKey, String> zSetOperations = this.redisKeyRedisOperations.boundZSetOps(redisKey);
        List<Error> errorCodeCount = convertErrors(code, zSetOperations.reverseRangeWithScores(offset, offset + limit - 1));
        if (errorCodeCount.size() == 0) {
            throw new ErrorsNotFoundException(40403, "service.errors.get.not.found", new Object[]{application, code});
        }
        return new Errors(errorCodeCount);
    }

    @Override
    public void saveErrors(String user, String application, Errors errors) {
        if (errors != null) {
            RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.ERROR).data(user).build();
            BoundZSetOperations<RedisKey, String> zSetOperations = this.redisKeyRedisOperations.boundZSetOps(redisKey);
            for (Error error : errors.getErrorCodeCount()) {
                zSetOperations.incrementScore(error.getCode(), error.getCount());
            }
        }
    }

    @Cacheable(value = "quotas",
            key = "#user.concat('-').concat(#application).concat('-').concat(#category)")
    public Quotas getQuotas(String user, String application, String category) throws QuotasNotFoundException {
        List<Quota> quotas = Lists.newLinkedList();
        RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.QUOTA)
                .category(StringUtils.hasText(category) ? category : RedisKey.REDIS_WILDCARD)
                .data(StringUtils.hasText(user) ? user : RedisKey.ANONYMOUS).build();
        Set<String> keys = stringRedisOperations.keys(redisKey.serialize());
        for (final String key : keys) {
            RedisKey rKey = RedisKey.deSerialize(key);
            Long hits = this.longRedisOperations.opsForValue().get(rKey);
            if (hits != null) {
                quotas.add(new Quota(rKey.getCategory(), hits));
            }
        }
        if (quotas.size() == 0) {
            throw new QuotasNotFoundException(40404, "service.quotas.get.not.found", new Object[]{application, category});
        }
        return new Quotas(quotas);
    }

    @Override
    public void saveQuotas(String user, String application, Quotas quotas) {
        if (quotas != null && quotas.getQuotas() != null) {
            for (Quota quota : quotas.getQuotas()) {
                RedisKey redisKey = new RedisKeyBuilder().key(application).type(KeyType.QUOTA).category(quota.getMethod()).data(user).build();
                this.longRedisOperations.opsForValue().increment(redisKey, quota.getHits());
            }
        }
    }

    @Cacheable(value = "notifications",
            key = "#user.concat('-').concat(#application).concat('-').concat(#category).concat('-').concat(#notificationId)")
    public Notifications getNotifications(String user, String application, String category, long notificationId) {
        // TODO possible optimization is to delete (by default + override) the accessed notifications (or just let them expire)
        Notifications notifications = new Notifications();
        RedisKey notificationCriteria = new RedisKeyBuilder().key(application).type(KeyType.NOTIFICATION)
                .category(StringUtils.hasText(category) ? category : RedisKey.REDIS_WILDCARD)
                .data(user).data((notificationId != -1) ? String.valueOf(notificationId) : RedisKey.REDIS_WILDCARD).build();
        Set<String> keys = stringRedisOperations.keys(notificationCriteria.serialize());
        for (final String key : keys) {
            RedisKey rKey = RedisKey.deSerialize(key);
            BoundZSetOperations<RedisKey, String> zSetOperations = this.redisKeyRedisOperations.boundZSetOps(rKey);
            Set<ZSetOperations.TypedTuple<String>> typedTuples = zSetOperations.reverseRangeWithScores(0, -1);
            Set<Notification> notificationSet = Sets.newHashSet();
            if (typedTuples != null) {
                for (ZSetOperations.TypedTuple<String> typedTuple : typedTuples) {
                    notificationSet.add(new Notification(Long.valueOf(rKey.getSecondData()), typedTuple.getScore()));
                }
            }
            if (notifications.getNotifiedService(rKey.getCategory()) == null) {
                notifications.addNotifiedService(new NotifiedService(rKey.getCategory(), notificationSet));
            } else {
                notifications.getNotifiedService(rKey.getCategory()).getNotifications().addAll(notificationSet);
            }
        }
        return notifications;
    }

    private List<Error> convertErrors(String code, Set<ZSetOperations.TypedTuple<String>> typedTuples) {
        List<Error> errorList = Lists.newArrayList();
        for (ZSetOperations.TypedTuple<String> typedTuple : typedTuples) {
            String errorCode = typedTuple.getValue();
            if (MatchingUtil.matchesWildcard(RedisKey.REDIS_WILDCARD, code, errorCode)) {
                errorList.add(new Error(errorCode, typedTuple.getScore()));
            }
        }
        return errorList;
    }

    private TrendedService convertTrend(String instanceName, String instanceIndex, String service, TrendType trendType,
                                        Set<ZSetOperations.TypedTuple<String>> typedTuples) {
        Multimap<TrendType, TrendPoint> trendPointMultimap = TreeMultimap.create();
        for (ZSetOperations.TypedTuple<String> tuple : typedTuples) {
            trendPointMultimap.put(trendType, new TrendPoint(new DateTime(Long.parseLong(tuple.getValue())), tuple.getScore()));
        }
        return new TrendedService(service, instanceName, instanceIndex, trendPointMultimap);
    }

    private Info getInfo(String keysCriteria) {
        Info info = new Info();
        Set<String> keys = stringRedisOperations.keys(keysCriteria);
        for (final String redisKey : keys) {
            RedisKey rKey = RedisKey.deSerialize(redisKey);
            switch (rKey.getType()) {
                case PERFORMANCE:
                    info.performanceData().put(rKey.getCategory(), new Instance(rKey.getInstanceName(), rKey.getInstanceIndex()));
                    break;
                case ERROR:
                    info.setErrorCodes(stringRedisOperations.opsForZSet().range(redisKey, 0L, -1L));
                    break;
                case NOTIFICATION:
                    info.notificationData().put(rKey.getCategory(), Long.valueOf(rKey.getSecondData()));
                    break;
                case QUOTA:
                    info.getQuotaData().add(rKey.getCategory());
                    break;
                case CACHE:
                    info.cacheData().put(rKey.getFirstData(), rKey.getSecondData());
                    break;
                default:
                    break;
            }
        }
        return info;
    }

    public void setStringRedisOperations(RedisOperations<String, String> stringRedisOperations) {
        this.stringRedisOperations = stringRedisOperations;
    }

    public void setBitmapRedisOperations(RedisOperations<String, BitSet> bitmapRedisOperations) {
        this.bitmapRedisOperations = bitmapRedisOperations;
    }

    public void setRedisKeyRedisOperations(RedisOperations<RedisKey, String> redisKeyRedisOperations) {
        this.redisKeyRedisOperations = redisKeyRedisOperations;
    }

    public void setLongRedisOperations(RedisOperations<RedisKey, Long> longRedisOperations) {
        this.longRedisOperations = longRedisOperations;
    }

    public void setRedisConfigManager(RedisConfigManager redisConfigManager) {
        this.redisConfigManager = redisConfigManager;
    }
}
